package com.hzlj.position.job.listener;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import com.fz.common.base.core.JsonUtils;
import com.fz.common.kafka.annotation.KafkaListenerRetry;
import com.google.common.collect.Lists;
import com.hzlj.position.config.config.PositionConfig;
import com.hzlj.position.locate.common.model.Position;
import com.hzlj.position.locate.service.PositionAddService;
import com.hzlj.sqjz.lc.common.dto.lcWc.LcWcQueryResultDTO;
import com.hzlj.sqjz.lc.service.LcWcService;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

import static com.hzlj.position.config.common.constant.Topic.POSITION_ADD;
import static com.hzlj.position.config.common.constant.Topic.POSITION_SYNC_LEAVING;


/**
 * @author lifh
 * @date 2023/3/17 11:29
 **/
@Slf4j
@Component
public class PositionAddListener {
    @Resource
    private PositionAddService positionAddService;
    @Resource
    private PositionConfig positionConfig;
    @Resource
    private LcWcService lcWcService;
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    private long startTime = System.currentTimeMillis();
    private final List<Position> positions = Lists.newArrayList();
    private final ReentrantLock lock = new ReentrantLock();

    @KafkaListenerRetry(enable = false)
    @KafkaListener(topics = {POSITION_ADD})
    public void positionAdd(ConsumerRecord<String, Position> record) {
        try {
            lock.lock();
            positions.add(record.value());
            //同步位置
            sync();
            //同步位置(如果是外出)
            syncPositionIfLeaving(record.value());
        } finally {
            lock.unlock();
        }
    }

    @Scheduled(cron = "0/1 * * * * ?")
    public void send() {
        try {
            lock.lock();
            sync();
        } finally {
            lock.unlock();
        }
    }


    private void sync() {
        //时间和数量的比较
        boolean needAdd = positions.size() >= positionConfig.getAddBatchSize()
                || (System.currentTimeMillis() - startTime) / 1000 >= positionConfig.getAddBatchSeconds();
        try {
            if (needAdd) {
                positionAddService.add(positions);
            }
        } catch (Exception e) {
            log.error("【kafka】【定位数据】批量新增异常,deviceCode-channel={}",
                    positions.stream().map(ee -> ee.getDeviceCode() + "-" + ee.getChannel()).collect(Collectors.joining(",")), e);
        } finally {
            if (needAdd) {
                startTime = System.currentTimeMillis();
                positions.clear();
            }
        }
    }

    private void syncPositionIfLeaving(Position position) {
        if (ObjectUtil.isEmpty(position.getLatitude())
                || ObjectUtil.isEmpty(position.getLongitude())) {
            return;
        }
        //外出时，同步位置
        if (!Boolean.TRUE.equals(positionConfig.getSyncPositionIfLeaving())) {
            return;
        }
        //是否有外出
        List<LcWcQueryResultDTO> lcs = lcWcService.listLcWc4Leaving(position.getJzId());
        if (ObjectUtil.isEmpty(lcs)) {
            return;
        }
        log.info("当前正在外出，需要同步位置信息:jzId={},xm={},positionId={},positionTime={},position={}",
                position.getJzId(), position.getXm(), position.getId(), position.getPositionTime(), position.getPosition());

        //leaving信息
        SyncPositionLeaving leaving = new SyncPositionLeaving();
        leaving.syncPosition = BeanUtil.copyProperties(position, SyncPosition.class);
        leaving.leavings = lcs.stream().map(e -> new Leaving()
                .setId(e.getId())
                .setLclx(e.getLclx())).collect(Collectors.toList());
        //发送通知
        kafkaTemplate.send(POSITION_SYNC_LEAVING, JsonUtils.obj2json(leaving));
    }

    @Data
    private static class SyncPositionLeaving {
        private List<Leaving> leavings;
        private SyncPosition syncPosition;
    }

    @Data
    @Accessors(chain = true)
    private static class Leaving {
        private String id;
        private String lclx;
    }

    @Data
    private static class SyncPosition {
        /**
         * 主键
         */
        private String id;
        /**
         * 机构ID全路径
         */
        private String jgCode;
        /**
         * 机构名称
         */
        private String jgMc;
        /**
         * 矫正ID(jz_jbxx)
         */
        private String jzId;
        /**
         * 矫正人员名称
         */
        private String xm;
        /**
         * 设备编号
         */
        private String deviceCode;
        /**
         * 定位渠道类型,dict-LOCATE_CHANNEL_TYPE
         */
        private String channelType;
        /**
         * 定位渠道,dict-LOCATE_CHANNEL
         */
        private String channel;
        /**
         * 经度
         */
        private String longitude;
        /**
         * 纬度
         */
        private String latitude;
        /**
         * 位置描述
         */
        private String position;
        /**
         * 定位时间-由设备产生
         */
        private Date positionTime;
        /**
         * 告警类型,dict-LOCATE_WARNING_TYPE
         */
        private String warningType;
        /**
         * 是否是告警记录
         */
        private Boolean warning;
    }

}
